package customized_input;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.net.URI;

public class CustomizedInput {
    private static final String INPUT_PATH = "hdfs://localhost:9000/recordread";
    private static final String OUTPUT_PATH = "hdfs://localhost:9000/recordput";

    public static void main(String[] args) throws Exception {
        //System.setProperty("HADOOP_USER_NAME", "collinsliu");
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(INPUT_PATH), conf);
        if (fs.exists(new Path(OUTPUT_PATH))) {
            fs.delete(new Path(OUTPUT_PATH), true);
        }
        // Job is a factory method, it will deliver necessary parameters to
        // corresponding conf object, also it will optimize its
        // construction logics
        Job myJob = Job.getInstance(conf, "RecordReaderMyJob");
        // set main class of job
        myJob.setJarByClass(CustomizedInput.class);
        // set mapper
        myJob.setMapperClass(MyMapper.class);
        myJob.setMapOutputKeyClass(LongWritable.class);
        myJob.setMapOutputValueClass(Text.class);
        // set reducer
        myJob.setReducerClass(MyReducer.class);
        myJob.setOutputKeyClass(Text.class);
        myJob.setOutputValueClass(LongWritable.class);
        // set partition
        myJob.setPartitionerClass(MyPartitioner.class);
        myJob.setNumReduceTasks(2);
        // set input format
        myJob.setInputFormatClass(MyInputFormat.class);
        FileInputFormat.addInputPath(myJob, new Path(INPUT_PATH));
        // set output format
        myJob.setOutputFormatClass(TextOutputFormat.class);
        FileOutputFormat.setOutputPath(myJob, new Path(OUTPUT_PATH));
        // submit job
        System.exit(myJob.waitForCompletion(true) ? 1 : 0);
    }
}
